package org.xcorpio.java8.chapter07;

import java.util.Spliterator;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

public class WordSpliterator {

    public static void main(String[] args) {
        final String sentence = "Hi, ,  hello  world.";
        Function<String, Integer> countWordsIteratively = (String s) -> {
            int counter = 0;
            boolean lastSpace = true;
            for (char c : s.toCharArray()) {
                if (Character.isWhitespace(c)) {
                    lastSpace = true;
                } else {
                    if (lastSpace) {
                        counter++;
                    }
                    lastSpace = false;
                }
            }
            return counter;
        };
        System.out.println(countWordsIteratively.apply(sentence));

        // parallelism is not supported
        WordCounter wordCounter = IntStream.range(0, sentence.length())
                .mapToObj(sentence::charAt)
                .parallel()
                .reduce(new WordCounter(0, true), WordCounter::accumulate, WordCounter::combine);
        System.out.println(wordCounter.getCounter());

        // parallelism is supported
        WordCounterSpliterator spliterator = new WordCounterSpliterator(sentence);
        Stream<Character> parallelStream = StreamSupport.stream(spliterator, true);
        WordCounter wordCounter2 = parallelStream.reduce(new WordCounter(0, true), WordCounter::accumulate,
                WordCounter::combine);
        System.out.println(wordCounter2.getCounter());

    }
}

class WordCounterSpliterator implements Spliterator<Character> {

    private final String string;
    private int currentChar = 0;

    public WordCounterSpliterator(String string) {
        this.string = string;
    }

    @Override
    public boolean tryAdvance(Consumer<? super Character> action) {
        // Consume the current character.
        action.accept(this.string.charAt(currentChar++));
        // Return true if there are further characters to be consumed.
        return currentChar < string.length();
    }

    @Override
    public Spliterator<Character> trySplit() {
        int currentSize = this.string.length() - this.currentChar;
        if (currentSize < 10) {
            /**
             * Return null to signal that the Stirng to be parsed is small
             * enough to be processed sequentially.
             */
            return null;
        }
        // Set the candidate's split position to be half of the String to be
        // parsed.
        for (int splitPos = currentSize / 2 + this.currentChar; splitPos < this.string.length(); splitPos++) {
            // Advance the split position until the next space.
            if (Character.isWhitespace(string.charAt(splitPos))) {
                /**
                 * Create a new WordCounterSpliterator parsing the String from
                 * the start to the split position.
                 */
                WordCounterSpliterator spliterator = new WordCounterSpliterator(
                        this.string.substring(currentChar, splitPos));
                // Set the start position of this WordCounterSpliterator to the
                // split position.
                this.currentChar = splitPos;
                return spliterator;
            }
        }
        return null;
    }

    @Override
    public long estimateSize() {
        return this.string.length() - currentChar;
    }

    @Override
    public int characteristics() {
        return ORDERED + SIZED + SUBSIZED + NONNULL + IMMUTABLE;
    }
}

/**
 * A class to count words while traversing a stream of Characters
 */
class WordCounter {
    private final int counter;
    private final boolean lastSpace;

    public WordCounter(int counter, boolean lastSpace) {
        this.counter = counter;
        this.lastSpace = lastSpace;
    }

    /**
     * The accumulate method traverses the Characters one by one as done by the
     * iterative algorithm.
     */
    public WordCounter accumulate(Character c) {
        if (Character.isWhitespace(c)) {
            return this.lastSpace ? this : new WordCounter(this.counter, true);
        } else {
            /**
             * Increase the word counter when the last character is a space and
             * the currently traversed one isn't
             */
            return this.lastSpace ? new WordCounter(this.counter + 1, false) : this;
        }
    }

    /**
     * Combine two WordCounters by summing their counters
     */
    public WordCounter combine(WordCounter wordCounter) {
        /**
         * Use only the sum of the counters so you don't care about lastSpace
         */
        return new WordCounter(this.counter + wordCounter.counter, wordCounter.lastSpace);
    }

    public int getCounter() {
        return counter;
    }
}
